# -*- encoding: utf-8 -*-

"""
------------------------------------------
@File       : life_game_v4.py
@Author     : maixiaochai
@Email      : maixiaochai@outlook.com
@CreatedOn  : 2021/7/23 10:30
------------------------------------------
    [ 生命游戏，Queue并发版 v3.2 ]
    与 v3版很相似，只是加了 count_neighbors函数I/O 的情况和相关的处理
    思路：
        在推进生命游戏时，提前创建数量固定的一组工作线程，
        令这组线程平行地处理当前这批 I/O任务，
        在处理完之后，等待下一批任务。
        如此，能减少很多资源消耗，也能减少频繁新建线程而带来的开销


    关系图：
        game_logic_thread有多个线程，以下为上一代产生下一代的推进过程，这个过程进行了5次,
        每次进行的时候in_queue和out_queue用的是全局的变量，中间 5个game_logic_thread线程也是一直存在的。
        或者可以这样理解，在生命游戏的5次推进中，整个流水线流水作业了5次。
        simulate_pipeline 是最辛苦的工人，堪称劳模，即使生产者也是消费者，干完前边工位的活儿，又去干后边工位的活儿。

                                         |--count_neighbors_thread_1 --|                  |--game_logic_thread_1 --|
                                         |  count_neighbors_thread_2   |                  |  game_logic_thread_2   |
        simulate_pipeline -> in_queue_ ->|  count_neighbors_thread_3   |-> logic_queue -> |  game_logic_thread_3   | -> out_queue_ -> simulate_pipeline
                                         |  count_neighbors_thread_4   |                  |  game_logic_thread_4   |
                                         |--count_neighbors_thread_5 --|                  |--game_logic_thread_5 --|
    优点（与v2相比）：
        1）使用内存少
        2）启动线程的开销小
        3）调试容易

    缺点:
        1）为了让代码更容易理解，必须使用 CloseableQueue和 StoppableWorker这样的类表示特质的类与工作线程，这会让程序变得复杂
        2）与并行度有关的参数（game_logic_thread函数的数量），必须依工作负载提前订好，而无法由系统根据工作量自动调整
        3）为了便于调试，需要在工作线程里手动捕获异常，并把它放入队列中，然后重新抛给主线程
        4）最严重的问题在需求发生变化的时候 —— 除了game_logic函数，现在 count_neighbors函数也需要做 I/O
"""
from queue import Queue
from threading import Thread
from time import sleep

from life_game_v1 import ALIVE, ColumnPrinter
from life_game_v2 import game_logic, LockingGrid


class SimulationError(Exception):
    """自定义错误类型"""
    pass


class ClosableQueue(Queue):
    SENTINEL = object()

    def close(self):
        self.put(self.SENTINEL)

    def __iter__(self):
        while True:
            item = self.get()

            # 在 try finally结构中，如果try中遇到了return，是会先执行finally中的代码，再执行try中的return
            try:
                if item is self.SENTINEL:
                    return  # Cause the thread to exit
                yield item

            finally:
                # get 不会让queue中的未完成数-1，必须配合task_done,才能让queue中的未完成的任务数-1
                # 这里用finally的原因，即使是try下边return了，task_done也会执行，保证队列内被处理的任务标记为已处理
                self.task_done()


class StoppableWorker(Thread):
    def __init__(self, func, in_queue, out_queue, **kwargs):
        super().__init__(**kwargs)
        self.func = func
        self.in_queue = in_queue
        self.out_queue = out_queue

    def run(self):
        for item in self.in_queue:
            result = self.func(item)
            self.out_queue.put(result)


def count_neighbors(y, x, get):
    """
        辅助函数，计算相邻单元格的状态
        计算顺序：从北方开始，逆时针转一圈
    """
    n_ = get(y - 1, x + 0)  # North
    ne = get(y - 1, x + 1)  # Northeast
    e_ = get(y + 0, x + 1)  # East
    se = get(y + 1, x + 1)  # Southeast
    s_ = get(y + 1, x + 0)  # South
    sw = get(y + 1, x - 1)  # Southwest
    w_ = get(y + 0, x - 1)  # West
    nw = get(y - 1, x - 1)  # Northwest

    neighbor_states = [n_, ne, e_, se, s_, sw, w_, nw]

    count = 0
    for state in neighbor_states:
        if state == ALIVE:
            count += 1

    sleep(0.1)
    return count


def count_neighbors_thread(item):
    y, x, state, get = item

    try:
        neighbors = count_neighbors(y, x, get)
    except Exception as e:
        neighbors = e

    return y, x, state, neighbors


def game_logic_thread(item):
    y, x, state, neighbors = item

    if isinstance(neighbors, Exception):
        next_state = neighbors

    else:
        try:
            next_state = game_logic(state, neighbors)
        except Exception as e:
            next_state = e

    return y, x, next_state


def simulate_pipeline(grid, in_queue, logic_queue, out_queue):
    for y in range(grid.height):
        for x in range(grid.width):
            state = grid.get(y, x)
            item = (y, x, state, grid.get)
            in_queue.put(item)  # fan out

    # join, 会在队列存在未完成任务时阻塞，等待队列无未完成任务，需要配合 task_done 使用
    in_queue.join()
    logic_queue.join()
    # A：这一步只是向out_queue中放入了一个 close的标记（而不是马上结束queue），当queue中按顺序识别到close标记的时候，会跳出阻塞式循环get
    out_queue.close()

    next_grid = LockingGrid(grid.height, grid.width)

    # 这里的使用，在于CloseableQueue中设计地比较巧妙，在遍历到close标记时，退出阻塞式get，并且还给queue最后一个任务打了一个task_done,完美
    for item in out_queue:  # fan in
        y, x, next_state = item

        if isinstance(next_state, Exception):
            raise SimulationError(y, x) from next_state
        next_grid.set(y, x, next_state)

    return next_grid


def main():
    """
        主要业务逻辑
    """
    in_queue_ = ClosableQueue()

    # 用于保存已经计算出周围有多少生命体的值
    logic_queue_ = ClosableQueue()
    out_queue_ = ClosableQueue()

    # 将 [ in_queue_ -> | count_neighbors_thread X 5 | -> logic_queue_ -> | game_logic_thread X 5 | -> out_queue_ ]
    # (这个不完整的)流水线建立起来
    threads = []
    for _ in range(5):
        thread = StoppableWorker(count_neighbors_thread, in_queue_, logic_queue_)
        thread.start()
        threads.append(thread)

    for _ in range(5):
        thread = StoppableWorker(game_logic_thread, logic_queue_, out_queue_)
        thread.start()
        threads.append(thread)

    grid = LockingGrid(5, 9)
    grid.set(0, 3, ALIVE)
    grid.set(1, 4, ALIVE)
    grid.set(2, 2, ALIVE)
    grid.set(2, 3, ALIVE)
    grid.set(2, 4, ALIVE)

    columns = ColumnPrinter()

    # 这里，simulate_pipeline 即充当了队列1之前的生产者，又充当了队列2的消费者,
    # 只不过是全部生产完全部东西放入in_queue_后，跑到最后(out_queue_)的后边"工位"，去充当消费者

    for i in range(5):
        columns.append(str(grid))
        grid = simulate_pipeline(grid, in_queue_, logic_queue_, out_queue_)

    # 其实print(columns)放这里也可以，但是不好理解，所以向后移了
    # 因为默认情况下（setDaemon=False），主线程结束之后，子线程会继续执行，直到子线程任务完成而结束
    # join 所完成的工作就是线程同步，即主线程任务结束之后，进入阻塞状态，一直等待其他的子线程执行结束之后，主线程再终止
    # print(columns)

    # in_queue_.close()/logic_queue.close()，放在这里，是因为前边声明游戏的整个流程是单线程的，已经完成了五次
    # 队列里什么也没有，但是这两个queue后边的消费者线程都被queue.get阻塞着，需要给他们发停止信号，
    # 让他它们结束get，停止在while下的不停地get，跳出while循环后，线程也结束了。

    # 为什么是5个结束标志呢？因为是都是五个消费者线程，每个消费者线程在get到一个结束标志后，便跳出不停get的while循环，
    # 不再get而消耗其它的停止信号，所以，5个不多不少，正好一个线程一个
    for _ in range(5):
        in_queue_.close()

    for _ in range(5):
        logic_queue_.close()

    # join 所完成的工作就是线程同步，即主线程任务结束之后，进入阻塞状态，一直等待其他的子线程执行结束之后，主线程再终止
    for thread in threads:
        thread.join()

    print(columns)


def demo():
    """
        在 try finally结构中，如果try中遇到了return，是会先执行finally中的代码，再执行try中的return
        out:
            2
            3
    """
    try:
        return 3

    finally:
        print("2")


if __name__ == '__main__':
    main()
